package com.crazymakercircle.completableFutureDemo;

import com.crazymakercircle.util.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;

import static com.crazymakercircle.util.ThreadUtil.sleepSeconds;

@Slf4j
public class CompletableFutureDemo {
    ExecutorService executorService = Executors.newFixedThreadPool(2);

    @Test
    public void thenApply() throws Exception {

        CompletableFuture<Void> cf = CompletableFuture.supplyAsync(() -> {
            try {
                //休眠2秒
                Thread.sleep(2000);
            } catch (Exception e) {
                e.printStackTrace();
            }
            log.info("supplyAsync");
            return "hello ";
        }, executorService).thenAccept(s -> {
            try {
                log.info("thenApply_test " + s + "world");
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        log.info("(-_-)");
        while (true) {
            if (cf.isDone()) {
                log.info("CompletedFuture...isDown");
                break;
            }
        }
    }

    //无返回值异步调用
    @Test
    public void runAsyncDemo() throws Exception {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            sleepSeconds(1);//模拟执行1秒
            log.info("run end ...");
        });

        //等待异步任务执行完成,现时等待2秒
        future.get(2, TimeUnit.SECONDS);
    }

    //有返回值异步调用
    @Test
    public void supplyAsyncDemo() throws Exception {
        CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
            long start = System.currentTimeMillis();
            sleepSeconds(1);//模拟执行1秒
            log.info("run end ...");
            return System.currentTimeMillis() - start;
        });

        //等待异步任务执行完成,现时等待2秒
        long time = future.get(2, TimeUnit.SECONDS);
        log.info("异步执行耗时（秒） = " + time / 1000);
    }

    @Test
    public void whenCompleteDemo() throws Exception {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() ->
        {
            sleepSeconds(1);//模拟执行1秒
            log.info("抛出异常！");
            //throw new RuntimeException("发生异常");
            //Print.tco("run end ...");
            return 100;
        });
        //设置执行完成后的回调钩子
        future.whenComplete(new BiConsumer<Integer, Throwable>() {
            @Override
            public void accept(Integer t, Throwable action) {
                log.info("执行完成！");
                log.info("ret:{}", t);
            }
        });
        //设置发生异常后的回调钩子
        future.exceptionally(t -> {
            log.info("执行失败！" + t.getMessage());
            return null;
        });
        Integer join = future.join();
        log.info("after join:{}", join);
        //TimeUnit.SECONDS.sleep(1000);
    }

    @Test
    public void handleDemo() throws Exception {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            sleepSeconds(1);//模拟执行1秒
            log.info("抛出异常！");
            // throw new RuntimeException("发生异常");
            //log.info("run end ...");
        });
        //设置执行完成后的回调钩子
        future.handle(new BiFunction<Void, Throwable, Void>() {
            @Override
            public Void apply(Void input, Throwable throwable) {
                if (throwable == null) {
                    log.info("没有发生异常！");

                } else {
                    log.info("sorry,发生了异常！");
                }
                return null;
            }
        });

        future.get();
    }

    //有返回值异步调用
    @Test
    public void threadPoolDemo() throws Exception {
        //业务线程池
        ThreadPoolExecutor pool = ThreadUtil.getMixedTargetThreadPool();
        CompletableFuture<Long> future = CompletableFuture.supplyAsync(() ->
        {
            log.info("run begin ...");
            long start = System.currentTimeMillis();
            sleepSeconds(1);//模拟执行1秒
            log.info("run end ...");
            return System.currentTimeMillis() - start;
        }, pool);

        //等待异步任务执行完成,现时等待2秒
        long time = future.get(2, TimeUnit.SECONDS);
        log.info("异步执行耗时（秒） = " + time / 1000);
    }

    @Test
    public void thenApplyDemo() throws Exception {
        CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
            long firstStep = 10L + 10L;
            log.info("firstStep outcome is " + firstStep);
            return firstStep;
        }).thenApplyAsync(new Function<Long, Long>() {
            @Override
            public Long apply(Long firstStepOutCome) {
                long secondStep = firstStepOutCome * 2;
                log.info("secondStep outcome is " + secondStep);
                return secondStep;
            }
        });

        long result = future.get();
        log.info(" future is " + future);
        log.info(" outcome is " + result);
    }

    @Test
    public void thenComposeDemo() throws Exception {
        CompletableFuture<Long> future = CompletableFuture.supplyAsync(new Supplier<Long>() {
            @Override
            public Long get() {
                long firstStep = 10L + 10L;
                log.info("firstStep outcome is " + firstStep);

                return firstStep;
            }
        }).thenCompose(new Function<Long, CompletionStage<Long>>() {
            @Override
            public CompletionStage<Long> apply(Long firstStepOutCome) {
                return CompletableFuture.supplyAsync(new Supplier<Long>() {
                    @Override
                    public Long get() {
                        long secondStep = firstStepOutCome * 2;
                        log.info("secondStep outcome is " + secondStep);
                        return secondStep;
                    }
                });
            }

        });
        long result = future.get();
        log.info(" outcome is " + result);
    }

    @Test
    public void thenCombineDemo() throws Exception {
        CompletableFuture<Integer> future1 =
                CompletableFuture.supplyAsync(() -> {
                    int firstStep = 10 + 10;
                    log.info("firstStep outcome is " + firstStep);
                    try {
                        TimeUnit.SECONDS.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return firstStep;
                });
        CompletableFuture<Integer> future2 =
                CompletableFuture.supplyAsync(() -> {
                    int secondStep = 10 + 10;
                    log.info("secondStep outcome is " + secondStep);
                    return secondStep;
                });
        CompletableFuture<Integer> future3 = future1.thenCombine(future2,
                (step1OutCome, step2OutCome) -> step1OutCome * step2OutCome);
        Integer result = future3.get();
        log.info(" outcome is " + result);
    }

    @Test
    public void applyToEitherDemo() throws Exception {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
            int firstStep = 10 + 10;
            log.info("firstStep outcome is " + firstStep);
            return firstStep;
        });
        CompletableFuture<Integer> future2 =
                CompletableFuture.supplyAsync(() -> {
                    int secondStep = 100 + 100;
                    log.info("secondStep outcome is " + secondStep);
                    return secondStep;
                });
        CompletableFuture<Integer> future3 = future1.applyToEither(future2,
                new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer eitherOutCome) {
                        return eitherOutCome;
                    }
                });
        Integer result = future3.get();
        log.info(" outcome is " + result);
    }

    @Test
    public void allOfDemo() {
        // supplyAsync
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
            log.info("模拟异步任务1");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 100;
        });
        // runAsync
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
            log.info("模拟异步任务2");
            return 101;
        });
        CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> {
            log.info("模拟异步任务3");
            return 103;
        });
        CompletableFuture<Integer> future4 = CompletableFuture.supplyAsync(() -> {
            log.info("模拟异步任务4");
            return 104;
        });

        CompletableFuture<Void> all = CompletableFuture.allOf(future1, future2, future3, future4);
        all.join();
        log.info("he");
    }

}
